package org.spdata.tsdb;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.bson.types.ObjectId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spdata.tsdb.client.DownSample;
import org.spdata.tsdb.dao.DpsDAO;
import org.spdata.tsdb.dao.MetricDAO;
import org.spdata.tsdb.dao.SnapshotDAO;
import org.spdata.tsdb.dto.DataPoint;
import org.spdata.tsdb.entity.DpsPack;
import org.spdata.tsdb.entity.Metric;
import org.spdata.tsdb.entity.Metric.ValueType;
import org.spdata.tsdb.entity.Snapshot;
import org.spdata.tsdb.entity.Snapshot.Type;
import org.spdata.tsdb.service.DataPointService;
import org.spdata.utils.ByteUtils;
import org.springframework.beans.factory.annotation.Autowired;


/**
 * 用数组缓存每个测点的采样数据,格式如下:
 * [{4个字节表示的metricId},tail0,tail1,{4个字节表示当前小时值},{8个字节表示首个采样值入库时间},{2个字节表示当前快照位置},采样时间(2 bytes)，采样值(4bytes)......]
 */

public class MemTable {
	static Logger logger = LoggerFactory.getLogger(MemTable.class);
	
	static final int head = 20;
	
	@Autowired
	DataPointService service;
	
	@Autowired
	DpsDAO dao;
	
	@Autowired
	MetricDAO metricDAO = null;
	
	@Autowired
	SnapshotDAO snapshotDAO = null;

	Map<Integer,byte[]> dpsMap = new ConcurrentHashMap<Integer,byte[]>();
	
	private int defaultBytesLenForDpsOfHour = 5120;
	
	private int defaultExtraBytesLenForDpsOfHour = 512;
	
	protected ConcurrentLinkedQueue<Integer> toSaveQueue =  new ConcurrentLinkedQueue<Integer>();
	
	private ConcurrentHashMap<Integer,Thread> writeFlags = new ConcurrentHashMap<Integer,Thread>();
	
	private boolean doingFullSnapshot = false;
	
	public int getDefaultBytesLenForDpsOfHour() {
		return defaultBytesLenForDpsOfHour;
	}
	
	public void setDefaultBytesLenForDpsOfHour(int defaultBytesLenForDpsOfHour) {
		this.defaultBytesLenForDpsOfHour = defaultBytesLenForDpsOfHour;
	}
	
	
	public int getDefaultExtraBytesLenForDpsOfHour() {
		return defaultExtraBytesLenForDpsOfHour;
	}
	
	public void setDefaultExtraBytesLenForDpsOfHour(int defaultExtraBytesLenForDpsOfHour) {
		this.defaultExtraBytesLenForDpsOfHour = defaultExtraBytesLenForDpsOfHour;
	}
	
	

	public SnapshotDAO getSnapshotDAO() {
		return snapshotDAO;
	}

	public void setSnapshotDAO(SnapshotDAO snapshotDAO) {
		this.snapshotDAO = snapshotDAO;
	}

	public DpsDAO getDao() {
		return dao;
	}
	
	public void setDao(DpsDAO dao) {
		this.dao = dao;
	}
	
	public DataPointService getService() {
		return service;
	}
	
	public void setService(DataPointService service) {
		this.service = service;
	}

	@PreDestroy
	public void destory() {
		//等待归档操作的执行
		while(toSaveQueue.size()>0) {
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
		//等待归档操作的完成
		try {
			Thread.sleep(100);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		//TODO:把无需归档的数据写到磁盘
		this.additionSnapshot();
		//清除缓存数据
		this.dpsMap.clear();
		this.timer.cancel();
		this.timer4Snapshot.cancel();
	}
	
	
	public void clear() {
		this.dpsMap.clear();
	}
	
	Timer timer =null;
	Timer timer4Snapshot = null;
	
	private void importSnapshots() {
		
		logger.info("正在导入快照");
		Date fullshotAt = null;
		//先导入全量快照
		List<Snapshot> snapshots = snapshotDAO.findLastFullSnapshot();
		if(snapshots==null || snapshots.size()==0)
			logger.info("没有需要导入的快照");
		else {
			for(Snapshot s:snapshots) {
				if(fullshotAt == null)
					fullshotAt = s.getTime();
				Map<String,Object> fragment = null;
				while((fragment=s.nextFramgment())!=null) {
					int metricId =(Integer) fragment.get("metricId");
					byte[] data = (byte[]) fragment.get("data");
					byte[] bytes = dpsMap.get(metricId);
					if(bytes==null) {
						bytes = new byte[this.defaultBytesLenForDpsOfHour];
						dpsMap.put(metricId, bytes);
					}else {
						logger.error("这里不该有数据");
					}
					System.arraycopy(data, 0, bytes, 0, data.length);
					this.writeTail(bytes, data.length);
					this.writeSnapshot(bytes, data.length);
				}
			}
	
			logger.info("正在导入增量快照");
			
			//再导入增量快照
			int batchSize = 20;
			List<Snapshot> additions = snapshotDAO.findAdditionSnapshots(fullshotAt,batchSize);
			for(;;) {
				for(Snapshot s:additions) {
					logger.info("导入增量快照数据，快照时间={}",s.getTime());
					Map<String,Object> fragment = null;
					while((fragment=s.nextFramgment())!=null) {
						int metricId =(Integer) fragment.get("metricId");
						
						int from = (Integer)fragment.get("from");
						byte[] data = (byte[]) fragment.get("data");
						byte[] bytes = dpsMap.get(metricId);
						if(bytes==null) {
							//按全量快照导入
							bytes = new byte[this.defaultBytesLenForDpsOfHour];
							dpsMap.put(metricId, bytes);
							System.arraycopy(data, 0, bytes, from, data.length);
							
						}else {
							if(from+data.length>bytes.length) {
								byte[] newArray = new byte[bytes.length+bytes.length*20/100];
								System.arraycopy(bytes, 0, newArray, 0, bytes.length);
								bytes = newArray;
								dpsMap.put(metricId, bytes);
							}
							try {
								System.arraycopy(data, 0, bytes, from, data.length);
							}catch(Exception e) {
								e.printStackTrace();
							}
						}
						this.writeTail(bytes, from+data.length);
						this.writeSnapshot(bytes, from+data.length);
						
						if(metricId==1) {
							long hour = this.readSampleHour(bytes);
							int tail = this.readTail(bytes);
							int second = ByteUtils.changeBytesToInt(bytes[tail-6],bytes[tail-5]);
							logger.info("已写入增量数据，time={},快照时间={}",new Date((hour*3600+second)*1000),s.getTime());
						}
					}
				}
				if(additions.size()==batchSize) {
					ObjectId after = additions.get(batchSize-1).getId();
					additions = snapshotDAO.nextPageSnapshots(after, batchSize);
				}else
					break;
			}
		}
		logger.info("快照导入结束");
	}
	
	private boolean importing = false;
	@PostConstruct
	public void init() {
		//加载快照数据
		importing = true;
		importSnapshots();
		importing = false;
		
		timer4Snapshot = new Timer();
		//写增量快照的定时器
		timer4Snapshot.schedule(new TimerTask() {
			@Override
			public void run() {
				//全量快照和初始化加载时不允许增量快照
				if(doingFullSnapshot || importing)
					return;
				if(snapshotDAO.existFullSnapshot())
					additionSnapshot();
				else
					fullSnapshot();
			}
		}, 30000, 20000);
		//定时检查是否有需要归档的serial
		timer = new Timer();
		timer.schedule(new TimerTask() {
			DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			@Override
			public void run() {
				int count= 0;
				Integer metricId = null;
				
				while((metricId = toSaveQueue.poll())!=null) {
					if(count==0) {
						doingFullSnapshot = true;
						logger.info("start to save dps in hour!");
					}
					//锁定id对应的 byte array
					lockMetricArray(metricId);
					try {
						byte[] bytes = dpsMap.get(metricId);
						DpsPack dps = readAsPack(bytes);
						try {
							boolean exists = dao.exists(dps.getId());
							if(!exists) {
								dao.create(dps);
								if(dps.getMetricId().intValue()==1) {
									System.out.println("压缩前:");
									System.out.println(dps.getDps());
									System.out.println("压缩后:");
									System.out.println(ByteUtils.changeBytesToHexStr(dps.toBytes()));
								}
							}
						}catch(Exception e) {
							String time = df.format(dps.getHour()*3600000);
							logger.error("failed to save dps,metricId={},hour={}",metricId,time,e);
						}
						copyExtraBytesToMainBytes(metricId);
						count++;
					}catch(Exception e) {
						logger.error("failed to save dps of metric[id={}]",metricId,e);
					}finally {
						unlockMetricArray(metricId);
					}
				}
				if(count>0) {
					logger.info("finished save dps in hour!count = {}",count);
					//如果有归档数据，则做一次全量快照
					fullSnapshot();
					doingFullSnapshot = false;
				}
			}
		}, 1000*2, 200);
	}
	
	private boolean lockMetricArray(Integer metricId) {
		Thread lockedBy = null;
		for(;;) {
			lockedBy = writeFlags.putIfAbsent(metricId,Thread.currentThread());
			if(lockedBy==null) {
				return true;
			}else {
				logger.debug("the data array of metric[id={}] is locked by {}",metricId,lockedBy);
			}
		}
	}
	
	private void unlockMetricArray(Integer metricId) {
		writeFlags.remove(metricId);
	}
	/**
	 * 全量快照
	 */
	long id = 0;
	protected void fullSnapshot() {
		try {
			logger.info("start to do full snapshot");
			Collection<Metric> metrics = service.getMetrics().values();
			Snapshot ss = new Snapshot();
			ss.setTime(new Date());
			ss.setIndex(0);
			ss.setType(Type.full);
			ss.setParent(null);
			for(Metric metric:metrics) {
				Integer id = metric.getId();
				this.lockMetricArray(id);
				try {
					//如果有可用扩展数组，则用扩展数组做快照
					byte[] bytes = dpsMap.get(0-id);
					if(!isAvailableArray(bytes)) {
						bytes = dpsMap.get(id);
					}
					if(bytes==null)
						continue;
					
					int len = this.readTail(bytes);
					if(ss.restDataSize() < (4+2+len)) {
						Snapshot saved = snapshotDAO.create(ss);
						logger.info("写入快照："+saved);
						Snapshot newPart = new Snapshot();
						newPart.setTime(saved.getTime());
						newPart.setIndex(saved.getIndex());
						newPart.setType(Type.part);
						newPart.setParent(saved.getParent()==null?saved.getId():saved.getParent());
						ss = newPart;
					}
					
					
					//写metricId
					ss.writeData(ByteUtils.changeIntTo4Bytes(id));
					ss.writeData(ByteUtils.changeIntTo2Bytes(0));
					//写data长度
					ss.writeData(ByteUtils.changeIntTo2Bytes(len));
					//写data
					ss.writeData(bytes,0,len);
					
					
					//修改最近快照的位置
					this.writeSnapshot(bytes, len);
				}catch(Exception e) {
					logger.error("failed to do snapshot of metric[id={}]",id,e);
				}finally {
					this.unlockMetricArray(id);
				}
			}
			if(ss.getData()!=null && ss.getData().length>0) {
				try {
					this.snapshotDAO.create(ss);
					logger.info("写入快照："+ss);
				} catch (Exception e) {
					logger.error("failed to save snapshot to mongo",e);
				}
			}
			logger.info("finish to do full snapshot");
			this.snapshotDAO.deleteTimeoutSnapshots(ss.getTime());
		}catch(Exception e) {
			logger.info("failed to do full snapshot",e);
		}
		
	}
	
	/**
	 * 增量快照
	 */
	protected void additionSnapshot() {
		logger.info("start to do addition snapshot");
		Snapshot ss = new Snapshot();
		ss.setTime(new Date());
		ss.setIndex(0);
		ss.setType(Type.addition);
		ss.setParent(null);
		
		Collection<Metric> metrics = service.getMetrics().values();
		for(Metric metric:metrics) {
			Integer id = metric.getId();
			this.lockMetricArray(id);
			try {
				//如果有可用扩展数组，则用扩展数组做快照
				byte[] bytes = dpsMap.get(0-id);
				if(!isAvailableArray(bytes)) {
					bytes = dpsMap.get(id);
				}
				if(bytes==null)
					continue;
				//写快照数据
				ss = writeAdditionSnapshot(ss,metric,bytes);
			}catch(Exception e) {
				logger.error("failed to do snapshot of metric[id={}]",id,e);
			}finally {
				this.unlockMetricArray(id);
			}
		}
		if(ss.getData()!=null && ss.getData().length>0) {
			try {
				this.snapshotDAO.create(ss);
				logger.info("写入快照："+ss);
			} catch (Exception e) {
				logger.error("failed to save snapshot to mongo",e);
			}
		}
		logger.info("finish to do addition snapshot");
	}
	
	private Snapshot writeAdditionSnapshot(Snapshot ss,Metric metric,byte[] bytes)throws Exception {
		int preSnapPos = this.readSnapshot(bytes);
		if(metric.getId().intValue()==1) {
			logger.info("上次快照截止位置:{}",preSnapPos);
		}
		int tail = this.readTail(bytes);
		if(tail>preSnapPos) {
			int len = tail - preSnapPos;
			if(ss.restDataSize() < (4+2+len)) {				
				Snapshot saved = this.snapshotDAO.create(ss);
				logger.info("写入快照："+saved);
				Snapshot newPart = new Snapshot();
				newPart.setTime(saved.getTime());
				newPart.setIndex(saved.getIndex());
				newPart.setType(Type.part);
				newPart.setParent(saved.getParent()==null?saved.getId():saved.getParent());
				ss = newPart;
			}
			//写metricId
			ss.writeData(ByteUtils.changeIntTo4Bytes(metric.getId()));
			//写本次快照起始位置
			ss.writeData(ByteUtils.changeIntTo2Bytes(preSnapPos));
			//写data长度
			ss.writeData(ByteUtils.changeIntTo2Bytes(len));
			//写data
			ss.writeData(bytes, preSnapPos, len);
			
			this.writeSnapshot(bytes, tail);
			if(metric.getId().intValue()==1) {
				logger.info("上次快照截止位置:{}",tail);
			}
			
		}
		return ss;
		
	}
	
	protected boolean isAvailableArray(byte[] bytes) {
		if(bytes==null)
			return false;
		
		int id = this.readTail(bytes);
		if(id>0) {
			return true;
		}else
			return false;
		
	}
	
	protected void copyExtraBytesToMainBytes(Integer metricId) {
		byte[] extraBytes =this.dpsMap.get(0-metricId);
		byte[] mainBytes = this.dpsMap.get(metricId);
		
		if(isAvailableArray(extraBytes)) {
			int tail = this.readTail(extraBytes);
			System.arraycopy(extraBytes, 0, mainBytes, 0, tail);
			Arrays.fill(extraBytes,(byte)0);
		}else {
			Arrays.fill(mainBytes,(byte)0);
		}
		
	}
	
	public void putDataPoint(DataPoint dp) {
		if(dp.getTime()==null) {
			dp.setTime(System.currentTimeMillis());
		}
		
		this.lockMetricArray(dp.getMetricId());
		try {
			byte[] dpsBytes = dpsMap.get(dp.getMetricId());
			if(dpsBytes==null) {
				dpsBytes = this.createEmptyByteArray();
				initBytes(dpsBytes,dp);
				dpsMap.put(dp.getMetricId(),dpsBytes);
			}
			this.writeDataPoint(dpsBytes, dp);
		}catch(Exception e) {
			logger.error("failed to write dp to cache ,dp={}",dp,e);
		}finally {
			this.unlockMetricArray(dp.getMetricId());
		}
		
	}
	
	public Map<Integer,byte[]> getDpsMap() {
		return this.dpsMap;
	}
	
	public DpsPack getDPS(int metricId) {
		byte[] bytes = this.dpsMap.get(metricId);
		return this.readAsPack(bytes);
	}
	
	public DpsPack getDPSFromExtra(int metricId) {
		byte[] bytes = this.dpsMap.get(0-metricId);
		if(bytes==null) {
			return null;
		}else
			return this.readAsPack(bytes);
	}
	
	public DpsPack getDPS(int metricId,DownSample sample) {
		byte[] bytes = this.dpsMap.get(metricId);
		if(bytes==null)
			return null;
		DpsPack dps = this.readAsPack(bytes);
		if(dps!=null)
			doSample(sample, dps);
		return dps;
	}
	public DpsPack getDPSFromExtra(int metricId,DownSample sample) {
		byte[] bytes = this.dpsMap.get(0-metricId);
		if(bytes==null)
			return null;
		DpsPack dps = this.readAsPack(bytes);
		if(dps!=null)
			doSample(sample, dps);
		return dps;
	}
	
	private void doSample(DownSample sample, DpsPack dps) {
		if(sample!=null){
			if(sample.getUnit().equals(TimeUnit.HOURS) || sample.getUnit().equals(TimeUnit.DAYS)){
				dps.doSummary();
				dps.getDps().clear();
				dps.setDps(null);
			}
		}
	}
	protected void initBytes(byte[] bytes,DataPoint dp) {		
		long firstSampleAtHour = dp.getTime()/(3600*1000);
		this.writeMetricId(bytes, dp.getMetricId());
		this.writeTail(bytes, 20);
		this.writeSampleHour(bytes, (int)firstSampleAtHour);
		this.writeFirstCacheTime(bytes, System.currentTimeMillis());
		this.writeSnapshot(bytes, 0);
	}
	
	/**
	 * 把采样值写入数组
	 * @param bytes
	 * @param dp
	 */
	protected void writeDataPoint(byte[] bytes,DataPoint dp) {
		long sampleAtHour = dp.getTime()/(3600*1000);
		//如果bytes被清空了，需要重新初始化
		int metricId = this.readMetricId(bytes);
		if(metricId==0) {
			this.initBytes(bytes, dp);
		}
		
		long firstSampleAtHour = (long)readSampleHour(bytes);
		if(sampleAtHour<firstSampleAtHour) {
			logger.error("sample hour little then first sample hour,this dp should be throw out,dp={}",dp );
		}else if(sampleAtHour>firstSampleAtHour) {
			if(sampleAtHour>firstSampleAtHour+1) {
				logger.debug("sample hour is too large then firest sample hour,write is too fast!");
			}
			this.writeDPToExtraBytes(dp);
		}else {
			bytes = this.doWriteDataPoint(1,bytes, dp);
		}
	}
	
	
	protected void writeDPToExtraBytes(DataPoint dp) {
		if(importing) {
			logger.warn("TSDB 正在导入未归档数据！");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return ;
		}
		Integer id = dp.getMetricId();
		byte[] bytes = this.dpsMap.get(0-id);
		if(bytes==null){
			bytes = this.createEmptyByteArray4Extra();
			Arrays.fill(bytes, (byte)0);
			dpsMap.put(0-id, bytes);
		}
		if(this.readMetricId(bytes)==0) {
			this.initBytes(bytes, dp);
			this.toSaveQueue.offer(id);
			
		}
		this.doWriteDataPoint(-1,bytes, dp);
	}
	
	protected byte[] createEmptyByteArray4Extra() {
		byte[] bytes = new byte[this.defaultExtraBytesLenForDpsOfHour];		
		return bytes;
	}
	
	protected byte[] createEmptyByteArray() {
		byte[] bytes = new byte[this.defaultBytesLenForDpsOfHour];		
		return bytes;
	}
	
	DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	private synchronized String convertSecondToString(long second){	
		return df.format(new Date(second));
	}
	
	
	protected byte[] doWriteDataPoint(int arrayType,byte[] bytes ,DataPoint dp) {
		int tail = this.readTail(bytes);		
		long second = (dp.getTime() % 3600000)/1000;
		
		if(tail>20) {
			int lastSecond = ByteUtils.changeBytesToInt(bytes[tail-6],bytes[tail-5]);
			int lastHour = this.readSampleHour(bytes);
			int lastTime = lastHour<<32 + lastSecond*1000;
			if(second<=lastSecond) {
				logger.warn("无法写入时序错乱的采样数据,lastSecond={},currentSecond={},current dp = {}",convertSecondToString(lastTime),convertSecondToString(dp.getTime()),dp.toString());
				return bytes;
			}
		}
		
		//数组长度不够时，扩张20%
		if(bytes.length-tail < 6) {
			byte[] newArray = new byte[bytes.length+bytes.length*20/100];
			System.arraycopy(bytes, 0, newArray, 0, tail);
			bytes = newArray;
			dpsMap.put(arrayType * dp.getMetricId(), bytes);
		}
		
		//write sample time
		byte[] secondBytes = ByteUtils.changeIntTo2Bytes((int)second);		
		bytes[tail++] = secondBytes[0];
		bytes[tail++] = secondBytes[1];
		
		//write sample value
		Number number = dp.getValue();
		if(number instanceof Double) {
			number = new Float(number.floatValue());
		}
		int value = changeNumberToInt(number);
		byte[] valueBytes = ByteUtils.changeIntTo4Bytes(value);
		bytes[tail++] = valueBytes[0];
		bytes[tail++] = valueBytes[1];
		bytes[tail++] = valueBytes[2];
		bytes[tail++] = valueBytes[3];
		
		//move tail
		this.writeTail(bytes, tail);
		
		return bytes;
	}
	
	public static Number getValueFromInt(int data,Metric metric){
		if(metric.getType().equals(ValueType.intValue))
			return data;
		else{			
			return Float.intBitsToFloat(data);
		}
	}
	
	
	protected DpsPack readAsPack(byte[] bytes) {
		int metricId = this.readMetricId(bytes);
		if(metricId==0)
			return null;
		
		long hour = (long)this.readSampleHour(bytes);
		Metric metric = service.getMetric(metricId);
		DpsPack dps = new DpsPack(hour,metricId,null);
		int position = head;
		int tail = this.readTail(bytes);
		
		Map<Integer,Number> values = new LinkedHashMap<Integer,Number>();
		while(position<tail) {
			int second = ByteUtils.changeBytesToInt(bytes[position],bytes[position+1]);
			position+=2;
			int intValue =  ByteUtils.changeBytesToInt(bytes[position],bytes[position+1],bytes[position+2],bytes[position+3]);
			position+=4;
			values.put(second, getValueFromInt(intValue,metric));
		}
		dps.setDps(values);
		return dps;
	}
	
	private int changeNumberToInt(Number number) {
		if(number instanceof Float) {
			Float f = (Float)number;
			return Float.floatToIntBits(f);
		}else if(number instanceof Integer) {
			return (Integer)number;
		}else 
			throw new IllegalArgumentException("i can only deal int or float value!");
	}
	
	protected void writeMetricId(byte[] bytes,int metricId) {
		byte[] idBytes = ByteUtils.changeIntTo4Bytes(metricId);
		bytes[0] = idBytes[0];
		bytes[1] = idBytes[1];
		bytes[2] = idBytes[2];
		bytes[3] = idBytes[3];
	}
	
	protected int readMetricId(byte[] bytes) {
		int id = ByteUtils.changeBytesToInt(bytes[0],bytes[1],bytes[2],bytes[3]);
		return id;
	}
	
	
	/**
	 * 向数组写入当前尾部位置
	 * @param bytes
	 * @param tail
	 */
	protected void writeTail(byte[] bytes,int tail) {
		byte[] tailBytes = ByteUtils.changeIntTo2Bytes(tail);
		bytes[4] = tailBytes[0];
		bytes[5] = tailBytes[1];
	}
	
	protected int readTail(byte[] bytes) {
		int tail = ByteUtils.changeBytesToInt(bytes[4],bytes[5]);
		return tail;
	}
	
	/**
	 * 向数组写采样小时
	 * @param bytes
	 * @param dp
	 */
	protected void writeSampleHour(byte[] bytes,int firstSampeAtHour) {
		byte[] bytesTime = ByteUtils.changeIntTo4Bytes(firstSampeAtHour);
		bytes[6] = bytesTime[0];
		bytes[7] = bytesTime[1];
		bytes[8] = bytesTime[2];
		bytes[9] = bytesTime[3];		
	}
	
	protected int readSampleHour(byte[] bytes) {
		int hour = ByteUtils.changeBytesToInt(bytes[6],bytes[7],bytes[8],bytes[9]);
		return hour;
	}
	
	/**
	 * 向数组写第一个采样点的缓存时间
	 * @param bytes
	 * @param dp
	 */
	protected void writeFirstCacheTime(byte[] bytes,long firstCacheAt) {
		byte[] bytesTime = ByteUtils.changeLongTo8Bytes(firstCacheAt);
		bytes[10] = bytesTime[0];
		bytes[11] = bytesTime[1];
		bytes[12] = bytesTime[2];
		bytes[13] = bytesTime[3];	
		bytes[14] = bytesTime[4];	
		bytes[15] = bytesTime[5];	
		bytes[16] = bytesTime[6];	
		bytes[17] = bytesTime[7];	
	}
	
	protected long readFirstCacheTime(byte[] bytes) {
		byte[] bs = new byte[8];
		System.arraycopy(bytes, 10, bs, 0, 8);
		long l = ByteUtils.changeBytesToLong(bs);
		return l;
	}
	
	protected void writeSnapshot(byte[] bytes,int position) {
		byte[] pBytes = ByteUtils.changeIntTo2Bytes(position);
		bytes[18] = pBytes[0];
		bytes[19] = pBytes[1];
	}
	
	protected int readSnapshot(byte[] bytes) {
		int p = ByteUtils.changeBytesToInt(bytes[18],bytes[19]);
		return p;
	}
	
	static AtomicInteger count = new AtomicInteger(0);
	static AtomicInteger errCount = new AtomicInteger(0);
	static class TestTask implements Runnable{
		MemTable tb = null;
		public TestTask(MemTable table){
			tb = table;
		}
		@Override
		public void run() {
			for(;;) {
				int metricId = random();
				boolean r = false;
				for(;;) {
					r = tb.lockMetricArray(metricId);
					if(r==true) {
						//成功执行一次
						count.incrementAndGet();
						tb.unlockMetricArray(metricId);
						break;
					}else {
						//errCount.incrementAndGet();
					}
				}
				
				
			}
		}
		
		Integer random() {
			return (int) Math.round((Math.random()*10000));
		}
	}
	
	public static void main(String[] args) {
		ConcurrentHashMap<Integer, Thread> workingFlags = new ConcurrentHashMap<Integer,Thread>();
		Executor executor = Executors.newFixedThreadPool(10);
		MemTable tb = new MemTable();
		for(int i=0;i<10;i++) {
			executor.execute(new TestTask(tb));
		}
		long t0 = System.currentTimeMillis();
		for(;;) {
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			long t1 = System.currentTimeMillis();
			if(t1-t0>=1000) {
				System.out.println(";count="+count.get()+";errCount="+errCount.get()+";size="+workingFlags.size());
				t0 = t1;
				count.set(0);
				errCount.set(0);
			}
			
		}
		/*
		int h = (220 & 0xFF00) >> 8;
		int l = 220 & 0x00FF;
		
		System.out.println(Integer.toHexString(220)+":"+Integer.toHexString(h)+";"+Integer.toHexString(l));
		
		byte hb = (byte)h;
		byte lb = (byte)l;		
		System.out.println(""+hb+":"+lb);
		l =  lb & 0xFF;
		System.out.println(""+l);
		
		h = (320 & 0xFF00) >> 8;
		l = 320 & 0x00FF;
		System.out.println(Integer.toHexString(320)+":"+Integer.toHexString(h)+";"+Integer.toHexString(l));
		*/
	}
}
